package com.springboot.kafka.consumer;

import com.springboot.kafka.producer.RecordBean;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

public class RecordDeserializer implements Deserializer<RecordBean> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public RecordBean deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }
        if (data.length < 8) {
            throw new SerializationException("反序列化错误");
        }
        ByteBuffer buffer = ByteBuffer.wrap(data);
        int nameLen, ageLen;
        String name = "";
        Integer age = 0;
        nameLen = buffer.getInt();
        byte[] nameBytes = new byte[nameLen];
        buffer.get(nameBytes);

        ageLen = buffer.getInt();
        byte[] ageBytes = new byte[ageLen];
        buffer.get(ageBytes);
        try {
            name = new String(nameBytes, "utf-8");
            age = Integer.parseInt(new String(ageBytes, "utf-8"));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        RecordBean recordBean = new RecordBean(name, age);
        return recordBean;
    }

    @Override
    public void close() {

    }
}
